package realtime.app.dws;/*
 *@ClassName : Dws_Trade_Province_TradeSum
 *@Description : $ {description}
 *@Author : NingYu
 *@Date : 2023-03-27 20:{MINUTE}
 */

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.bean.TradeProvinceBean;
import realtime.func.AsyncDIMFunction;
import realtime.util.ClickHouseUtil;
import realtime.util.DateFormatUtil;
import realtime.util.MyKafkaUtil;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

public class Dws_Trade_Province_TradeSum {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_province_tradesum_0327_02";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    if (s != null) {
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        collector.collect(jsonObject);
                    }
                } catch (Exception e) {
                    System.out.println(s + " : 无法转换为json格式");
                }
            }
        });

        //
        SingleOutputStreamOperator<JSONObject> jsonWithWMDS = jsonDS
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("create_time");
                            }
                        }));

        //
        SingleOutputStreamOperator<TradeProvinceBean> beanDS = jsonWithWMDS.map(new RichMapFunction<JSONObject, TradeProvinceBean>() {

            @Override
            public TradeProvinceBean map(JSONObject jsonObject) throws Exception {

                String orderId = jsonObject.getString("order_id");
                String userId = jsonObject.getString("user_id");

                return TradeProvinceBean.builder()
                        .provinceId(jsonObject.getString("province_id"))
                        .finalAmount(jsonObject.getDouble("final_amount"))
                        .orderSet(new HashSet<String>(Collections.singleton(orderId)))
                        .userSet(new HashSet<String>(Collections.singleton(userId)))
                        .build();
            }
        });

        //
        SingleOutputStreamOperator<TradeProvinceBean> reduceDS = beanDS
                .keyBy(tradeProvinceBean -> tradeProvinceBean.getProvinceId())
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TradeProvinceBean>() {
                    @Override
                    public TradeProvinceBean reduce(TradeProvinceBean t1, TradeProvinceBean t2) throws Exception {
                        t1.getOrderSet().addAll(t2.getOrderSet());
                        t1.getUserSet().addAll(t2.getUserSet());
                        t1.setFinalAmount(t1.getFinalAmount() + t2.getFinalAmount());

                        return t1;
                    }
                }, new WindowFunction<TradeProvinceBean, TradeProvinceBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TradeProvinceBean> iterable, Collector<TradeProvinceBean> collector) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(timeWindow.getStart());
                        String edt = DateFormatUtil.toYmdHms(timeWindow.getEnd());

                        for (TradeProvinceBean bean : iterable) {
                            bean.setStt(stt);
                            bean.setEdt(edt);
                            bean.setPersonCount((long) (bean.getUserSet().size()));
                            bean.setOrderCount((long) (bean.getOrderSet().size()));
                            bean.setTs(System.currentTimeMillis());

                            collector.collect(bean);
                        }
                    }
                });

        SingleOutputStreamOperator<TradeProvinceBean> resultDS = AsyncDataStream.unorderedWait(
                reduceDS
                , new AsyncDIMFunction<TradeProvinceBean>("DIM_BASE_PROVINCE") {
                    @Override
                    public void join(TradeProvinceBean input, JSONObject dimInfo) {
                        input.setProvinceName(dimInfo.getString("NAME"));

                    }

                    @Override
                    public String getKey(TradeProvinceBean input) {
                        return input.getProvinceId();
                    }
                }
                , 60
                , TimeUnit.SECONDS
        );

        //
        resultDS.print("Dws_Trade_Province_TradeSum>>>>>>>>>>>>>\n");
        resultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_province_tradesum_window values(?,?,?,?,?,?,?,?)"));

        env.execute();

    }
}
